package sgt.test;

import ltq.util.*;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

public class CountBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Map<String,Integer> map;
    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;
    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
//        this.collector=collector;
        map = new HashMap<>();
        conn= DatabaseUtil.getConnection();
    }

    @Override
    public void execute(Tuple input) {

        //"img_src","title","price","deal","shop","location"
//        String img_src = input.getString(0);
//        String title = input.getString(1);
//        Integer price = Integer.parseInt(input.getString(2));
        String sdealNum = input.getString(3);
        Integer dealNum;
        if (sdealNum.equals("")){
            dealNum=0;
        }else {
            dealNum=Integer.parseInt(sdealNum);
        }

        String shop = input.getString(4);
//        String location = input.getString(5);


        if (map.containsKey(shop)) {
            Integer tmpNum = map.get(shop);
            tmpNum+=dealNum;
            map.put(shop, tmpNum);
//            System.out.println("当前结果：店铺："+shop+",成交量："+tmpNum);
        } else {
            map.put(shop, dealNum);
//            System.out.println("当前结果：店铺："+shop+",成交量："+dealNum);
        }
        refreshData(shop,map.get(shop));





    }


    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
    public void refreshData(String shop,int deal){

        try {
            ps= conn.prepareStatement("select * from lipstick where shop=?");
            ps.setObject(1,shop);
            rs= ps.executeQuery();
            if (rs.next()){
                ps= conn.prepareStatement("update lipstick set deal=? where shop=?");
                ps.setObject(1,deal);
//                System.out.println("-----num------"+num);
                ps.setObject(2,shop);
                ps.executeUpdate();
            }
            else {
                ps= conn.prepareStatement("insert into lipstick(deal,shop) values (?,?)");
                ps.setObject(1,deal);
                ps.setObject(2,shop);
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }

    }
}
